package cloud.xiguapi.ubas.analysis.hotitems;

import cloud.xiguapi.ubas.analysis.hotitems.model.*;
import cloud.xiguapi.ubas.model.HotItemsViewCount;
import cloud.xiguapi.ubas.model.UserBehavior;
import cloud.xiguapi.ubas.utils.JacksonUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

import static cloud.xiguapi.ubas.constant.JobNameConstant.FILE_HOT_ITEMS;

/**
 * 热门商品统计分析类
 * - 基于文件
 * - 离线分析
 *
 * @author 大大大西西瓜皮🍉
 * date: 2021-5-16 下午 11:42
 * desc:
 */
@Component
public class FileHotItemsAnalysis {

    private final StreamExecutionEnvironment env;

    @Autowired
    public FileHotItemsAnalysis(StreamExecutionEnvironment env) {
        this.env = env;
    }

    @Async("FileHotItemsAnalysis")
    public void analysis(String path) throws Exception {
        env.setParallelism(1);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        DataStreamSource<String> inputStream = env.readTextFile(path);

        // 转换并分配时间戳、水印
        SingleOutputStreamOperator<UserBehavior> userBehaviorStream = inputStream.map(line -> {
            String[] fields = line.split(",");
            return UserBehavior.builder()
                    .userId(Long.valueOf(fields[0]))
                    .itemId(Long.valueOf(fields[1]))
                    .categoryId(Long.valueOf(fields[2]))
                    .behavior(fields[3])
                    .timestamp(Long.valueOf(fields[4]))
                    .build();
        }).assignTimestampsAndWatermarks(new UserBehaviorAscendingTimestampExtractor());

        // 分组开窗聚合, 获得每个窗口内各个商品的热门度, 即count值
        SingleOutputStreamOperator<HotItemsViewCount> windowAggregateDataStream = userBehaviorStream
                // 筛选pv行为
                .filter(data -> "pv".equals(data.getBehavior()))
                // 按商品ID分组
                .keyBy("itemId")
                // 打开滑窗, 间隔1h, 步长5min
                .timeWindow(Time.hours(1L), Time.minutes(10L))
                .aggregate(new FileHotItemsCountAggregate(), new FileWindowHotItemsCountResult());

        // 收集同一窗口的所有商品count数据, 后排序, 输出top N
        DataStream<TopNHotItemsResult> resultDataStream = windowAggregateDataStream
                // 按照窗口分组
                .keyBy("windowEnd")
                // 用自定义处理函数排序取前10
                .process(new TopNHotItems(5));

        // sink到文件
        // resultDataStream.addSink(StreamingFileSink
        //         .forRowFormat(new Path("X:\\output\\ubas\\file-hotitems"), new SimpleStringEncoder())
        //         .build());

        // sink到ES
        // resultDataStream.addSink(ElasticsearchSinkUtils.<TopNHotItemsResult>build(((element, ctx, indexer) ->
        //         // indexer.add(Requests.indexRequest()
        //         //         .index("file-hotitems")
        //         //         .id(element.getWindowEnd())
        //         //         .source(element.getDetails()))
        //
        // {
        //     Map<String, String> source = new HashMap<>();
        //     source.put("id", element.getWindowEnd());
        //     source.put("details", JacksonUtils.obj2json(element.getDetails()));
        //
        //     IndexRequest request = Requests.indexRequest();
        //     request.index(FILE_HOT_ITEMS.getJobName())
        //             .source(source);
        //     indexer.add(request);
        // }
        // )));

        // sink mysql
        resultDataStream.print();
        resultDataStream.addSink(new FileHotItemsMySqlSink());

        env.execute(FILE_HOT_ITEMS.getJobName());
    }

    public static class FileHotItemsMySqlSink extends RichSinkFunction<TopNHotItemsResult> {

        private transient Connection connection;

        private transient PreparedStatement clearStatement;

        private transient PreparedStatement selectStatement;

        private transient PreparedStatement insertStatement;

        private transient PreparedStatement updateStatement;

        private transient boolean isFirst = true;

        @Override
        public void open(Configuration parameters) throws Exception {
            connection = DriverManager.getConnection("jdbc:mysql://localhost:3306/ubas?serverTimezone=UTC", "root", "1015");
            clearStatement = connection.prepareStatement("TRUNCATE TABLE file_hot_items");
            selectStatement = connection.prepareStatement("SELECT 1 FROM file_hot_items WHERE window_end = ? LIMIT 1");
            insertStatement = connection.prepareStatement("INSERT INTO file_hot_items(window_end, details) VALUES(?, ?)");
            updateStatement = connection.prepareStatement("UPDATE file_hot_items SET details = ? WHERE window_end = ?");
        }

        @Override
        public void close() throws Exception {
            updateStatement.close();
            selectStatement.close();
            clearStatement.close();
            insertStatement.close();
            connection.close();
        }

        @Override
        public void invoke(TopNHotItemsResult value, SinkFunction.Context context) throws Exception {
            if (isFirst) {
                clearStatement.execute();
                isFirst = false;
                return;
            }

            selectStatement.setString(1, value.getWindowEnd());
            boolean next = selectStatement.executeQuery().next();

            if (next) {
                updateStatement.setString(1, JacksonUtils.obj2json(value.getDetails()));
                updateStatement.setString(2, value.getWindowEnd());
                updateStatement.execute();
            } else {
                insertStatement.setString(1, value.getWindowEnd());
                insertStatement.setString(2, JacksonUtils.obj2json(value.getDetails()));
                insertStatement.execute();
            }
        }
    }
}
